package demo.android.com.mvpdemo.RxJava;

/**
 * Created by Vya on 2018/1/28.
 */

class ObservableSchedulers<T> extends Observable<T> {

    final Observable<T> source;
    final Schedulers schedulers;
    public ObservableSchedulers(Observable<T> source, Schedulers schedulers) {

        this.source = source;
        this.schedulers = schedulers;
    }

    @Override
    protected void subscribeActual(Observer<T> observer) {
        schedulers.scheduleDirect(new SchedulerTask(observer));
    }

    private class SchedulerTask implements Runnable{
        final  Observer<T> observer;
        public SchedulerTask( Observer<T> observer) {
            this.observer = observer;
        }

        @Override
        public void run() {
            //这里的调用会执行上层的subscribe
            source.subscribe(observer);
        }
    }
}
